#include "../../core/pch.h"
#include "../../pi/binarysemaphore.h"
#include "../utils/utils.h"

/* Struct for worker address. */
static struct sockaddr_in worker_addr;

/* For worker socket. */
static int worker_sock;

/* Send to all slaves END_OF_COMPUTATION packet.
   After receiving this packet slaves should kill
   themselves. */

static void
kill_slaves(struct info_table *shm)
{
	BYTE msg;
	for (int i = 0; i < shm->count; i++)
	{
		struct sockaddr_in work_addr;
		FILL_SOCKADDR_IN_TXT(work_addr, AF_INET, 
			shm->worker_port[i], shm->ip[i]);

		int work_sock = socket(AF_INET, SOCK_STREAM, 0);
		if (work_sock == SOCKET_ERROR)
		{
			DEBUGS("sender: socket error for worker (%s)", strerror(errno));
			exit(SOCKET_ERROR);
		}

		int ret = connect(work_sock, (struct sockaddr*) &work_addr, sizeof(work_addr));
		if (ret == CONNECT_ERROR)
		{
			DEBUGS("sender: connect error to worker (%s)", strerror(errno));
			close(work_sock);
			exit(CONNECT_ERROR);
		}

		msg = END_OF_COMPUTATION;
		send(work_sock, &msg, sizeof(BYTE), 0);
		close(work_sock);
	}
}

/* Creates new socket and connects it
   to worker. */

static NET_STATUS
create_socket_and_connect(struct info_table *shm, int &work_sock, 
	struct sockaddr_in &work_addr, int slave_num)
{
	close(work_sock);

	/* Create socket. */
	work_sock = socket(AF_INET, SOCK_STREAM, 0);
	if (work_sock == SOCKET_ERROR)
	{
		DEBUGS("sender: socket error for worker (%s)", strerror(errno));
		return SOCKET_ERR;
	}

	/* Try to connect to slave. */
	int ret = connect(work_sock, (struct sockaddr*) &work_addr, sizeof(work_addr));
	if (ret == CONNECT_ERROR)
	{
		DEBUGS("sender: connect error to worker (%s)", strerror(errno));
		shm->alive[slave_num] = DEAD;
		close(work_sock);
		return CONNECT_ERR;
	}

	return NET_OK;
}

/* Get first not sent prime number. */

static ULL
get_first_not_sent(FILE *primes_file, struct info_table *shm, int slave_num)
{
	/** Structure for primes record. */
	struct prime_str prim;

	/* Get prime from file. */
	prim = PF_get_first_not_sent(primes_file);
	PF_set_sent(prim.number, primes_file);
	shm->number[slave_num] = prim.number;

	return prim.value;
}

/* Main sender function which is looking for
   idle hosts and sending them factor requests. */

int sender(int segment_id, int sem_id, struct info_table *shm, FILE *primes_file,
	int &receiver_pid, int &logger_pid)
{
	DEBUGS("sender: running");

	BYTE msg;
	while (true)
	{
		BinarySemaphore::Decrement(sem_id);

		if (shm->n == 1)
			break;

		for (int i = 0; i < shm->count; i++)
		{
			FILL_SOCKADDR_IN_TXT(worker_addr, AF_INET, shm->worker_port[i], 
				shm->ip[i]);

			if (shm->idle[i] && shm->alive[i])
			{
				DEBUGS("sender: connecting to idle computer (%s, %d)", 
					shm->ip[i], shm->port[i]);

				NET_STATUS stat = create_socket_and_connect(shm, worker_sock, worker_addr, i); 
				if (stat != NET_OK) continue;

				ULL prime_to_check = get_first_not_sent(primes_file, shm, i);

				DEBUGS("sender: send number %llu", prime_to_check);

				msg = FACTOR_REQUEST;
				send(worker_sock, &msg, sizeof(char), 0);
				send(worker_sock, &shm->n, sizeof(ULL), 0);
				send(worker_sock, &prime_to_check, sizeof(ULL), 0);
				send(worker_sock, &shm->number[i], sizeof(UINT), 0);

				DEBUGS("sender: after sending number");

				shm->idle[i] = BUSY;

				close(worker_sock);
			}
		}

		BinarySemaphore::Increment(sem_id);
	}

	DEBUGS("sender: exiting");

	kill_slaves(shm);

	/* Kill other master processes. */
	kill(receiver_pid, SIGKILL);
	kill(logger_pid, SIGKILL);

	return STATUS_OK;
}

